package org.nbict.iot.task.realtime;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
import org.apache.storm.mongodb.common.SimpleQueryFilterCreator;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Fields;
import org.bson.Document;
import org.nbict.iot.PropertyUtil;
import org.nbict.iot.task.realtime.filter.AuthFilter;
import org.nbict.iot.task.realtime.filter.UploadFilter;
import org.nbict.iot.task.realtime.func.GisResolver;
import org.nbict.iot.task.realtime.func.PacketJsonProject;
import org.nbict.iot.task.realtime.func.PacketResolver;
import org.nbict.iot.task.realtime.map.GisDeviceStatusMap;
import org.nbict.iot.task.realtime.state.Comparer;
import org.nbict.iot.task.realtime.state.Timely;
import org.nbict.iot.task.realtime.state.TimelyFactory;
import org.nbict.iot.task.realtime.state.TimelyReplacer;
import org.nbict.iot.trident.kafka.KafkaSpout4ParsedTools;

/**
 * Created by songseven on 18/7/2.
 */
public class GisExactlyTrident {
    
    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {

        //设置kafka消费策略
        KafkaSpoutConfig.FirstPollOffsetStrategy strategy = KafkaSpoutConfig
                .FirstPollOffsetStrategy.EARLIEST;
        if (args.length > 1) {
            strategy = KafkaSpoutConfig.FirstPollOffsetStrategy.valueOf
                    (args[1]); //strategy name
        }

        String url = PropertyUtil.getProperty("mongo.url");
        String database = PropertyUtil.getProperty("mongo.database");
        String username = PropertyUtil.getProperty("mongo.username");
        String password = PropertyUtil.getProperty("mongo.password");
        String admin = PropertyUtil.getProperty("mongo.admin.name");
        String collectionName = "vehicle_map";

        TridentTopology topology = new TridentTopology();

        KafkaSpoutConfig spoutConfig = KafkaSpout4ParsedTools.newKafkaSpoutConfig
                ("192.168.1.83:9092", "transit-.*-simple", // bootstrap server, 192.168.1.83:9092
                        "iot-car-group-gis-update", strategy); //group id
        KafkaTridentSpoutOpaque kafkaSpout = KafkaSpout4ParsedTools
                .newKafkaTridentSpoutOpaque(spoutConfig);

        Stream stream = topology.newStream("iot-car-parsed-4-gis", kafkaSpout);

        Fields packetFields = new Fields("car_packet_header",
                "car_packet_body");
        Fields outputFields = new Fields("car_vin", "action", "timestamp",
                "upload");

        Stream stage = stream.each(new Fields("platform_id", "packet"), new PacketJsonProject(packetFields), packetFields)
                .each(new Fields("platform_id", "car_packet_header", "car_packet_body"), new PacketResolver(), outputFields)
                .project(new Fields("platform_id", "car_vin", "action", "timestamp", "upload")); //去除packet元组

        //实时上报信息流
        Stream uploadStream = stage.each(outputFields, new UploadFilter())
                .partitionBy(new Fields("car_vin"));

        //登入登出流
        Stream authStream = stage.each(outputFields, new AuthFilter());

        //地图流
        Timely.Options options = new Timely.Options()
                .withUrl(url)
                .withDatabase(database)
                .withUsername(username)
                .withPassword(password)
                .withAdmin(admin)
                .withCollectionName(collectionName);
        Fields persistFields = new Fields("car_vin", "timestamp", "status", "charged_state", "lon", "lat", "alarm_level");

        //更新整车信息及地理坐标
        uploadStream.each(
                new Fields("platform_id", "car_vin", "timestamp", "upload"),
                new GisResolver(), new Fields("status", "charged_state", "lon", "lat", "alarm_level"))
                .partitionPersist(
                        new TimelyFactory(
                                options,
                                new Comparer() { //持久化
                                    @Override
                                    public Document compareAndSet(TridentTuple tuple, Document replaced) {
                                        if (replaced == null)
                                            return replaced; //未注册车辆
                                        Long timestamp = tuple.getLong(1); //timestamp
                                        Long carTimestamp = replaced.getLong("timestamp");
                                        Long gisTimestamp = replaced.getLong("gis_timestamp");
                                        Long alarmTimestamp = replaced.getLong("alarm_timestamp");
                                        if (gisTimestamp == null || gisTimestamp < timestamp) {
                                            replaced.put("lon", tuple.get(4));
                                            replaced.put("lat", tuple.get(5));
                                            replaced.put("gis_timestamp", timestamp);
                                        }
                                        if (carTimestamp == null || carTimestamp < timestamp) {
                                            replaced.put("status", tuple.getString(2));
                                            replaced.put("charged_state", tuple.getString(3));
                                            replaced.put("timestamp", timestamp);
                                        }
                                        if (alarmTimestamp == null || alarmTimestamp < timestamp) {
                                            replaced.put("alarm_level", tuple.getString(6));
                                            replaced.put("alarm_timestamp", timestamp);
                                        }
                                        return replaced;
                                    }
                                }, new SimpleQueryFilterCreator().withField("car_vin")), //持久化查询字段
                        persistFields,
                        new TimelyReplacer(), new Fields());

        authStream.map(new GisDeviceStatusMap(), new Fields("car_vin", "timestamp", "device_status"))
                .partitionPersist(
                        new TimelyFactory(
                                options,
                                new Comparer() {
                                    @Override
                                    public Document compareAndSet(TridentTuple tuple, Document replaced) {
                                        if (replaced == null)
                                            return replaced; //未注册车辆
                                        Long timestamp = tuple.getLong(1);
                                        Long deviceTimestamp = replaced.getLong("device_timestamp");
                                        if (deviceTimestamp == null || deviceTimestamp < timestamp) {
                                            replaced.put("device_status", tuple.get(2));
                                            replaced.put("device_timestamp", timestamp);
                                        }
                                        return replaced;
                                    }
                                }, new SimpleQueryFilterCreator().withField("car_vin")), //持久化查询字段)
                        new Fields("car_vin", "timestamp", "device_status"),
                        new TimelyReplacer(), new Fields() ); //记录登入登出;

        Config config = new Config();

        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, topology.build());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("iot-car-gis", config, topology.build());
        }
    }

}
